package com.raylu.day09;

import com.raylu.utils.ClickEvent;
import com.raylu.utils.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

// 触发器
public class Example1trigger {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        // 最大延迟时间设置为0
                        WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                            @Override
                            public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                return element.ts;
                            }
                        })
                )
                .filter(r -> r.username.equals("Mary"))
                .keyBy(r -> r.username)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 触发器触发的是什么操作？
                // 触发的是后面的ProcessWindowFunction中的process函数
                .trigger(new MyTrigger())
                .process(new ProcessWindowFunction<ClickEvent, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<ClickEvent> elements, Collector<String> out) throws Exception {
                        out.collect("用户" + s + "在窗口" + new Timestamp(context.window().getStart()) + "~" +
                                "" + new Timestamp(context.window().getEnd()) + "中的pv是：" + elements.spliterator().getExactSizeIfKnown());
                    }
                })
                .print();

        env.execute();
    }

    public static class MyTrigger extends Trigger<ClickEvent, TimeWindow> {
        // 窗口中每来一条数据，触发执行一次
        @Override
        public TriggerResult onElement(ClickEvent element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            // 状态变量，用来标志到达的时间是否是窗口的第一个事件
            // 作用域是当前窗口
            ValueState<Boolean> isFirstEvent = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("is-first", Types.BOOLEAN));

            if (isFirstEvent.value() == null) {
                // 第一条数据到来，注册一个第一条数据时间戳后面的整数秒的定时器
                long nextSec = element.ts + 1000L - element.ts % 1000L;
                ctx.registerEventTimeTimer(nextSec); // 注册的是onEventTime定时器
                isFirstEvent.update(true);
            }
            return TriggerResult.CONTINUE;
        }

        // 处理时间到达`time`时，触发执行
        // 定时器
        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        // 水位线到达`time`时，触发执行
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            if (time < window.getEnd()) {
                if (time + 1000L < window.getEnd()) {
                    // 注册time后面的整数秒的定时器onEventTime
                    ctx.registerEventTimeTimer(time + 1000L);
                }
                // 触发窗口的计算（也就是process方法的执行）
                return TriggerResult.FIRE;
            }
            return TriggerResult.CONTINUE;
        }

        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            // 单例
            ValueState<Boolean> isFirstEvent = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("is-first", Types.BOOLEAN));
            isFirstEvent.clear();
        }
    }
}
